package org.zjt.util;

import com.alibaba.fastjson.JSONObject;
import io.netty.channel.ChannelFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zjt.config.RegisterConfig;
import org.zjt.domain.InvokeBean;
import org.zjt.domain.RegisteNodeData;
import org.zjt.domain.RpcMsgProto;
import org.zjt.netty.client.NettyClient;


import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.Comparator;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

import static org.zjt.util.RpcContext.LoadBalanceProtocol.ROTAION;

/**
 * Created by Administrator on 2017/4/12.
 */
public class InvokerInvocationHandler implements InvocationHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(InvokerInvocationHandler.class);

    private final InvokeBean<?> invoker;

    public InvokerInvocationHandler(InvokeBean<?> handler) {
        this.invoker = handler;
    }


    /**
     * method:调用的方法名   proxy:被代理的对象  args：调用方法传递来的参数
     */
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        /** method:调用的方法名   proxy:被代理的对象  args：调用方法传递来的参数*/
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        if (method.getDeclaringClass() == Object.class) {
            return method.invoke(invoker, args);
        }
        if ("toString".equals(methodName) && parameterTypes.length == 0) {
            return invoker.toString();
        }
        if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
            return invoker.hashCode();
        }
        if ("equals".equals(methodName) && parameterTypes.length == 1) {
            return invoker.equals(args[0]);
        }
        /**
         * 执行远程调用： return invoker.invoke(new RpcInvocation(method, args)).recreate(); */
        LOGGER.info(invoker.getClassName() + "\t" + JSONObject.toJSONString(proxy) + "\t" + JSONObject.toJSONString(method) + "\t" + JSONObject.toJSONString(args));
        Set<RegisteNodeData> hostodes = RegisterConfig.REGISTER_ROUTE_MAP.get(invoker.getClassName());
        if (hostodes == null)
            throw new Exception("无可用服务");
        RegisteNodeData serviceNodeData = handlerLoadBalance(hostodes);
        if (serviceNodeData == null)
            throw new Exception("无可用服务");
        String v1 = String.format("%s:%s", serviceNodeData.getHost(), serviceNodeData.getPort());
        ChannelFuture cf ;
        if (!RpcContext.RPC_NETTY_CLIENTS.containsKey(v1)) {
            cf = new NettyClient().connect(serviceNodeData.getPort(), serviceNodeData.getHost());
            RpcContext.RPC_NETTY_CLIENTS.put(v1, cf);
        } else {
            cf = RpcContext.RPC_NETTY_CLIENTS.get(v1);
        }
        RpcMsgProto.msg msg = InvokeMethodExecuter.createRpcMsgProto(serviceNodeData.getClassName(), method, args);
        String tranceId = String.format("%s:%s:%s", msg.getClassName(), msg.getMethodName(), msg.getParam());
        CountDownLatch countDownLatch = new CountDownLatch(1);
        RpcContext.RPC_TRAMCE_STATUS.put(tranceId, countDownLatch);
        serviceNodeData.initSendState();
        cf.channel().writeAndFlush(msg);
        //cf.channel().closeFuture().sync(); 关闭异步阻塞
        countDownLatch.await();
        RpcContext.RPC_TRAMCE_STATUS.remove(tranceId);
        serviceNodeData.updateReceiveState();
        RpcMsgProto.msg result = RpcContext.RPC_TRAMCE_MAP.remove(tranceId);
        if (result == null)
            throw new Exception("服务调用失败");
        if (result.getResultClassName() != null && !result.getResultClassName().equals("")) {
            Class c = InvokeMethodExecuter.handleSingleClss(result.getResultClassName());
            return JSONObject.parseObject(result.getResultJson(), c);
        }
        return null;
    }


    public static RegisteNodeData handlerLoadBalance(Set<RegisteNodeData> hostNodes){
        RpcContext.LoadBalanceProtocol loadBalanceProtocol = RpcContext.getLoadBalanceProtocol();
        if (hostNodes.size() < 2){
            Optional<RegisteNodeData> first = hostNodes.stream().findFirst();
            return first.isPresent() ? first.get() : null ;
        }
        //RADOM("随机访问", 1), REACTIME("最小时延", 2), ROTAION("轮询", 3);
        switch (loadBalanceProtocol){
            case RADOM:
                Optional<RegisteNodeData> any = hostNodes.stream().findAny();
                return any.isPresent() ? any.get() : null ;
            case REACTIME:
                Optional<RegisteNodeData> min = hostNodes.stream().min((o1, o2) ->
                        (o1.getCallTime() == null || o2.getCallTime() == null) ? o1.getCallTime() == null ? -1 : 1 : (o1.getCallTime().get() <= o2.getCallTime().get() ? -1 : 1));
                return min.isPresent() ? min.get() : null ;
            case ROTAION:
                Optional<RegisteNodeData> any1 = hostNodes.stream()
                        .filter(a ->a != RpcContext.currentServiceNode).findAny();
                return any1.isPresent() ? any1.get() : null ;
            default : return null;
        }

    }

}
